// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved.
// See License.txt in the project root for license information.
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading;
using System.Threading.Actors;
using System.Threading.Tasks;

namespace ActorStressTests
{
    class CommunicationTests
    {
        // Test ping-ponging via pubsub
        public static void PingPongPubSub(string fabricAddress, bool useGateway, int numIterations)
        {
            Console.WriteLine("Starting PingPongPubSub({0})...", numIterations);

            var pingActor = new PingPongActor("ping");
            var pongActor = new PingPongActor("pong");

            using (var pingClient = pingActor.DeployAndConnect(fabricAddress, useGateway, 3))
            using (var pongClient = pongActor.DeployAndConnect(fabricAddress, useGateway, 3))
            {
                // Acquire the executing assembly and ship it to both actors
                var asm = Assembly.GetExecutingAssembly();
                var name = asm.ManifestModule.Name;
                var location = asm.Location;
                var bytes = File.ReadAllBytes(location);

                pingClient.AddAssembly(name, bytes);
                pongClient.AddAssembly(name, bytes);

                // Instruct the actors to subscribe to each other
                pingClient.CallMethod("Initialize", new object[] { "fabric:/actor/adhoc/pong" });
                pongClient.CallMethod("Initialize", new object[] { "fabric:/actor/adhoc/ping" });

                TaskCompletionSource<bool> tcs = new TaskCompletionSource<bool>();
                pingClient.Subscribe("Status", new PingPongObserver(tcs));

                pingClient.CallMethod("StartPingPong", new object[] { numIterations });

                tcs.Task.Wait();
            }

            Console.WriteLine("PingPongPubSub({0}) completed successfully", numIterations);
        }

        // Test ping-ponging via "fire-and-forget" method calls
        public static void PingPongFNF(string fabricAddress, bool useGateway, int numIterations)
        {
            Console.WriteLine("Starting PingPongFNF({0})...", numIterations);

            var pingActor = new PingPongActor("ping");
            var pongActor = new PingPongActor("pong");

            using (var pingClient = pingActor.DeployAndConnect(fabricAddress, useGateway, 3))
            using (var pongClient = pongActor.DeployAndConnect(fabricAddress, useGateway, 3))
            {
                // Acquire the executing assembly and ship it to both actors
                var asm = Assembly.GetExecutingAssembly();
                var name = asm.ManifestModule.Name;
                var location = asm.Location;
                var bytes = File.ReadAllBytes(location);

                pingClient.AddAssembly(name, bytes);
                pongClient.AddAssembly(name, bytes);

                // Perform initialization if it has not been done already
                pingClient.CallMethod("Initialize", new object[] { "fabric:/actor/adhoc/pong" });
                pongClient.CallMethod("Initialize", new object[] { "fabric:/actor/adhoc/ping" });

                TaskCompletionSource<bool> tcs = new TaskCompletionSource<bool>();
                pingClient.Subscribe("Status", new PingPongObserver(tcs));

                pingClient.CallMethod("PingMethodPost", new object[] { numIterations+1 }); // So that we get even numIterations

                tcs.Task.Wait();
            }

            Console.WriteLine("PingPongFNF({0}) completed successfully", numIterations);
        }

        // Test ping-ponging via continuation-based result processing
        public static void PingPongAsync(string fabricAddress, bool useGateway, int numIterations)
        {
            Console.WriteLine("Starting PingPongAsync({0})...", numIterations);

            var pingActor = new PingPongActor("ping");
            var pongActor = new PingPongActor("pong");

            using (var pingClient = pingActor.DeployAndConnect(fabricAddress, useGateway, 3))
            using (var pongClient = pongActor.DeployAndConnect(fabricAddress, useGateway, 3))
            {
                // Acquire the executing assembly and ship it to both actors
                var asm = Assembly.GetExecutingAssembly();
                var name = asm.ManifestModule.Name;
                var location = asm.Location;
                var bytes = File.ReadAllBytes(location);

                pingClient.AddAssembly(name, bytes);
                pongClient.AddAssembly(name, bytes);

                // Perform initialization if it has not been done already
                pingClient.CallMethod("Initialize", new object[] { "fabric:/actor/adhoc/pong" });
                pongClient.CallMethod("Initialize", new object[] { "fabric:/actor/adhoc/ping" });

                TaskCompletionSource<bool> tcs = new TaskCompletionSource<bool>();
                pingClient.Subscribe("Status", new PingPongObserver(tcs));

                pingClient.CallMethod("StartAsyncPinging", new object[] { numIterations + 1 }); // so we get even iterations

                tcs.Task.Wait();

                pingClient.CallMethod("BogusAsyncCall", new object[0]);
            }

            Console.WriteLine("PingPongAsync({0}) completed successfully", numIterations);
        }

        // Test a slightly more convoluted downcounter example
        public static void Downcounter(string fabricAddress, bool useGateway, int numIterations)
        {
            Console.WriteLine("Starting Downcounter({0})...", numIterations);

            var dcActor = new PingPongActor("downcounter");
            var dcClientActor = new PingPongActor("dcClient");

            using (var dcClient = dcActor.DeployAndConnect(fabricAddress, useGateway, 3))
            using (var dcClientClient = dcClientActor.DeployAndConnect(fabricAddress, useGateway, 3))
            {
                // Acquire the executing assembly and ship it to both actors
                var asm = Assembly.GetExecutingAssembly();
                var name = asm.ManifestModule.Name;
                var location = asm.Location;
                var bytes = File.ReadAllBytes(location);

                dcClient.AddAssembly(name, bytes);
                dcClientClient.AddAssembly(name, bytes);

                // Do the necessary subscriptions
                TaskCompletionSource<bool> tcs = new TaskCompletionSource<bool>();
                dcClientClient.Subscribe("Update", new DowncountingObserver(tcs, numIterations));

                // Kicks off the downcounting
                dcClientClient.CallMethod("InitializeDCClient", new object[] { numIterations, "fabric:/actor/adhoc/downcounter" });

                tcs.Task.Wait();
            }

            Console.WriteLine("Downcounter({0}) completed successfully", numIterations);
        }

        // Test a slightly more convoluted downcounter example
        public static void DowncounterStateless(string fabricAddress, bool useGateway, int numIterations)
        {
            Console.WriteLine("Starting DowncounterStateless({0})...", numIterations);

            FabricActorClient.CreateStatelessActor(fabricAddress, "fabric:/actor/stateless/downcounter", StatelessActorStateTypes.InMemory, StatelessActorPersonalities.Empty);
            FabricActorClient.CreateStatelessActor(fabricAddress, "fabric:/actor/stateless/dcClient", StatelessActorStateTypes.InMemory, StatelessActorPersonalities.Empty);

            Uri fabricUri = new Uri("net.tcp://" + fabricAddress);
            using (var dcClient = new FabricActorClient( fabricUri, new Uri("fabric:/actor/stateless/downcounter"), useGateway))
            using (var dcClientClient = new FabricActorClient( fabricUri, new Uri("fabric:/actor/stateless/dcClient"), useGateway))
            {
                // Acquire the executing assembly and ship it to both actors
                var asm = Assembly.GetExecutingAssembly();
                var name = asm.ManifestModule.Name;
                var location = asm.Location;
                var bytes = File.ReadAllBytes(location);

                dcClient.AddAssembly(name, bytes);
                dcClientClient.AddAssembly(name, bytes);

                // Do the necessary subscriptions
                TaskCompletionSource<bool> tcs = new TaskCompletionSource<bool>();
                dcClientClient.Subscribe("Update", new DowncountingObserver(tcs, numIterations));

                // Kicks off the downcounting
                dcClientClient.CallMethod("InitializeDCClient", new object[] { numIterations, "fabric:/actor/stateless/downcounter" });

                tcs.Task.Wait();
            }

            Console.WriteLine("Downcounter({0}) completed successfully", numIterations);
        }


        public static void ActorLifespanManagement(string fabricAddress, bool useGateway, int numActors)
        {
            Console.WriteLine("Starting ActorLifespanManagement({0})...", numActors);
            var actorCreator = new PingPongActor("actorCreator");
            using (var acClient = actorCreator.DeployAndConnect(fabricAddress, useGateway, 3))
            {
                // Acquire the executing assembly and ship it to both actors
                var asm = Assembly.GetExecutingAssembly();
                var name = asm.ManifestModule.Name;
                var location = asm.Location;
                var bytes = File.ReadAllBytes(location);

                acClient.AddAssembly(name, bytes);

                acClient.CallMethod("ManageActors", new object[] {numActors});
            }

            Console.WriteLine("ActorLifespanManagement completed successfully");
        }

        public static void PublishSubscribeDeadlockAvoidanceTest(String fabricAddress, bool useGateway)
        {
            Console.WriteLine("Starting Publish/Subscribe deadlock avoidance test...");

            FabricActorClient.CreateStatelessActor(fabricAddress, "fabric:/actor/stateless/echoServer", StatelessActorStateTypes.InMemory, StatelessActorPersonalities.Empty);

            // The idea: Respond to a publish/subscribe message without deadlocking our thread that listens to sockets.
            // Have an IObserver<ActorEvent> that calls an actor method.  Ensure that doesn't cause a deadlock.  (It used to.)
            Uri fabricUri = new Uri("net.tcp://" + fabricAddress);
            using (var simpleClient = new FabricActorClient(fabricUri, new Uri("fabric:/actor/stateless/echoServer"), useGateway))
            {
                Object ret = null;
                IObserver<ActorEvent> observerThatCallsActorMethod = new EchoObserver(actorEvent => ret = simpleClient.CallMethod("ReturnOne", new Object[0]));

                using (simpleClient.Subscribe("EchoEvent", observerThatCallsActorMethod))
                {
                    // Add the assembly containing our echo actor
                    Assembly echoActorAssembly = typeof(EchoActor).Assembly;
                    byte[] assemblyBytes = File.ReadAllBytes(echoActorAssembly.Location);
                    simpleClient.AddAssembly(echoActorAssembly.ManifestModule.Name, assemblyBytes);

                    // Call a method that publishes an event
                    simpleClient.CallMethod("EchoStringToPublish", new Object[] { "Hi there" });

                    // Now ensure that the ActorRawTcpClient ReaderLoop isn't blocked.  Do this by calling another method.
                    Console.WriteLine("  If we hang here, the deadlock isn't fixed.");
                    simpleClient.CallMethod("ReturnOne", new Object[0]);
                    Console.WriteLine("  We're good.");

                    // "ret" is set asynchronously so we may need to wait a bit
                    var waiter = Task.Factory.StartNew(() =>
                        {
                            while (ret == null) Thread.Sleep(10);
                        });
                    var timedOut = !waiter.Wait(1000);
                    if (timedOut)
                        throw new Exception("Expected \"ret\" to have its value set within a second of EchoStringToPublish completing");

                    // Ensure that the observer wrote to our captured variable
                    if (ret == null || (ret.GetType() == typeof(Int64) && (long)ret != 1))
                        throw new Exception(String.Format("Publish/Subscribe deadlock avoidance test expected that our observer's callback called a method, got a " +
                                                          "valid response and wrote it to a local captured in a closure.  Expected 1 (as an Int32, but will tentatively accept Int64).  Got: {0}", ret));
                }
            }
            Console.WriteLine("Deadlock avoided.");
        }
    }

    internal class PingPongObserver : IObserver<ActorEvent>
    {
        TaskCompletionSource<bool> m_tcs;
        public PingPongObserver(TaskCompletionSource<bool> tcs)
        {
            m_tcs = tcs;
        }

        public void OnCompleted()
        {
        }

        public void OnError(Exception error)
        {
            throw new NotImplementedException();
        }

        public void OnNext(ActorEvent value)
        {
            var payload = value.GetTypedPayload<int>();
            if (payload <= 0)
            {
                m_tcs.SetResult(true);
            }
            else if ( (payload % 100) == 0)
            {
                Console.WriteLine("PingPongs remaining: {0}", payload);
            }
        }
    }

    // Completes TCS once the count reaches 0 and the TOTAL hit count reaches initialCount*2.
    // This is because two events should be released for each new count value: one from
    // the "UpdateHandler" event handler, one from the "SignalReturnHandler".
    internal class DowncountingObserver : IObserver<ActorEvent>
    {
        private TaskCompletionSource<bool> m_tcs;
        private int m_initialCount;

        public DowncountingObserver(TaskCompletionSource<bool> tcs, int initialCount)
        {
            m_tcs = tcs;
            m_initialCount = initialCount;
        }

        public void OnCompleted()
        {
        }

        public void OnError(Exception error)
        {
            throw new NotImplementedException();
        }

        public void OnNext(ActorEvent value)
        {
            var tuple = value.GetTypedPayload<Tuple<int,int>>();
            var newCount = tuple.Item1;
            var hitCount = tuple.Item2;

            if (newCount == 0 && hitCount == m_initialCount * 2)
            {
                m_tcs.SetResult(true);
            }
            else if ((newCount % 100) == 0)
            {
                Console.WriteLine("Latest count: {0}, hitcount: {1}", newCount, hitCount);
            }
        }
    }

    public class PingPongActor : FabricAdHocActorBase
    {
        public PingPongActor(string name)
            : base(name)
        {
        }

        // Subscribes, and stores the other actor's name
        [ActorMethod]
        public static object Initialize(IActorState state, object[] args)
        {
            object temp;
            // Avoid double-initializing
            if (!state.TryGet("_initialized", out temp))
            {
                var otherActorName = (string)args[0];
                state.Set("otherActorName", otherActorName);
                bool amIPing = otherActorName.Contains("pong");
                state.Set("MyName", amIPing ? "Ping" : "Pong");
                var proxy = state.GetActorProxy(otherActorName);
                if (amIPing)
                    proxy["Pong"].Subscribe("PongHandler");
                else
                    proxy["Ping"].Subscribe("PingHandler");
                state.Set("_initialized", true);
            }
            return true;
        }

        //
        // PingPonging via message publishing
        //

        // Starts pubsub-based ping-ponging.
        [ActorMethod]
        public static object StartPingPong(IActorState state, object[] args)
        {
            var initialCount = (int)args[0];
            state.Publish("Ping", initialCount);
            return initialCount;
        }

        // Handles "Ping" events
        [ActorMethod]
        public static object PingHandler(IActorState state, object[] args)
        {
            var ae = args[0] as ActorEvent;
            var payload = ae.GetTypedPayload<int>();
            state.Publish("Pong", payload);
            return null;
        }

        // Handles "Pong" events
        [ActorMethod]
        public static object PongHandler(IActorState state, object[] args)
        {
            var ae = args[0] as ActorEvent;
            var payload = ae.GetTypedPayload<int>();
            state.Publish("Status", payload);
            if (payload > 0)
            {
                state.Publish("Ping", payload - 1);
            }
            return null;
        }

        //
        // PingPonging via fire-and-forget method calls
        //

        // "Proactive" PingPong, instead of "Reactive" PingPong
        [ActorMethod]
        public static object PingMethodPost(dynamic state, dynamic[] args)
        {
            int remainingCount = args[0];
            state.Publish("Status", remainingCount - 1);
            if (remainingCount > 0)
            {
                string otherActorName = state.otherActorName;
                IActorProxy proxy = state.GetActorProxy(otherActorName);
                proxy.Command("PingMethodPost", new object[] { remainingCount - 1 });
            }
            return null;
        }

        //
        // PingPonging via async method calls
        //

        // The Ping method called in our async-method-based pingpong test.
        [ActorMethod]
        public static object PingMethodAsync(dynamic state, dynamic[] args)
        {
            int remainingCount = args[0];
            //Console.WriteLine("PingMethodAsync: See {0}", remainingCount);
            return remainingCount - 1;
        }

        // Initiates the PingPonging 
        [ActorMethod]
        public static object StartAsyncPinging(dynamic state, dynamic[] args)
        {
            int remainingCount = args[0];
            string otherActorName = state.otherActorName;
            IActorProxy proxy = state.GetActorProxy(otherActorName);
            var promise = proxy.Request("PingMethodAsync", new object[] { remainingCount }); 
            promise.ContinueWith("AsyncResultHandler");
            return null;
        }

        // The handler for async method call returns
        [ActorMethod]
        public static object AsyncResultHandler(dynamic state, dynamic[] args)
        {
            var promise = args[0];
            int currCount = promise.GetTypedResult<int>();
            state.Publish("Status", currCount);
            if (currCount > 0)
            {
                var proxy = state.GetActorProxy(state.otherActorName);
                proxy.PingMethodAsync(currCount - 1).ContinueWith("AsyncResultHandler");
            }
            return null;
        }

        // Just for fun, let's call a bogus actor method, hook up a continuation, and (externally) verify that the hiccup
        // shows up in our ETW log.
        [ActorMethod]
        public static object BogusAsyncCall(dynamic state, dynamic[] args)
        {
            string otherActorName = state.otherActorName;
            var proxy = state.GetActorProxy(otherActorName);
            var promise = proxy.SomeBogusMethod(1, 2, 3);
            promise.ContinueWith("AsyncResultHandler");
            return null;
        }

        //
        // Actor methods for Downcounter example
        //

        [ActorMethod]
        public static object InitializeDowncounter(dynamic state, dynamic[] args)
        {
            int initialCount = args[0];
            state.currentCount = initialCount;
            return initialCount;
        }

        [ActorMethod]
        public static object Signal(dynamic state, dynamic[] args)
        {
            if (state.currentCount > 0)
            {
                state.currentCount -= 1;
                state.Publish("Update", (int)state.currentCount);
            }
            return (int)state.currentCount;
        }

        [ActorMethod]
        public static object InitializeDCClient(dynamic state, dynamic[] args)
        {
            int initialCount = args[0];
            string dcActorName = (string)args[1];
            state.dcActorName = dcActorName;
            var proxy = state.GetActorProxy(dcActorName);
            proxy.InitializeDowncounter(initialCount);
            if (!state.Contains("_subscribed"))
            {
                proxy["Update"].Subscribe("UpdateHandler");
                state._subscribed = true;
            }

            state.totalHitCount = 0;

            // Kick off the process
            proxy.Signal().ContinueWith("SignalReturnHandler");

            return initialCount;
        }

        [ActorMethod]
        public static object UpdateHandler(dynamic state, dynamic[] args)
        {
            ActorEvent ae = args[0];
            var updatedCount = ae.GetTypedPayload<int>();
            state.totalHitCount++; // Allows us to test that state is properly updated/protected in event handler
            state.Publish("Update", Tuple.Create(updatedCount, (int) state.totalHitCount));
            return null;
        }

        [ActorMethod]
        public static object SignalReturnHandler(dynamic state, dynamic[] args)
        {
            var promise = args[0];
            int updatedCount = promise.GetTypedResult<int>();
            state.totalHitCount++; // Allows us to test that state is properly updated/protected in promise continuation
            state.Publish("Update", Tuple.Create(updatedCount, (int) state.totalHitCount));

            if (updatedCount > 0)
            {
                var proxy = state.GetActorProxy((string)state.dcActorName);
                proxy.Signal().ContinueWith("SignalReturnHandler");
            }

            return null;
        }

        [ActorMethod]
        public static object ManageActors(dynamic state, dynamic[] args)
        {
            int numActors = args[0];
            for (int i = 0; i < numActors; i++)
            {
                string actorName = "fabric:/actor/adhoc/lifetimeActor" + i;
                state.CreateActor(actorName, "applicationType=fabric:/actor/adhoc;serviceType=EmptyActorReplica;numReplicas=3");
                var proxy = state.GetActorProxy(actorName);
                string actorId = proxy.Ping().GetTypedResult<string>();
                if (!actorId.Equals(actorName))
                    throw new Exception(String.Format("Expected ping result ({0}) to be the same as actor name ({1})", actorId, actorName));
                state.DeleteActor(actorName);
            }

            return null;
        }
    }

    public class EchoActor : FabricAdHocActorBase
    {
        public EchoActor(String name) : base(name)
        {
        }

        [ActorMethod]
        public static Object EchoStringToPublish(IActorState state, Object[] args)
        {
            String str = (String)args[0];
            state.Publish("EchoEvent", str);
            return null;
        }

        [ActorMethod]
        public static Object ReturnOne(IActorState state, Object[] args)
        {
            return 1;
        }
    }

    public class EchoObserver : IObserver<ActorEvent>
    {
        private Action<ActorEvent> _func;

        public EchoObserver(Action<ActorEvent> func)
        {
            _func = func;
        }

        public void OnCompleted()
        {
        }

        public void OnError(Exception error)
        {
            throw new NotImplementedException();
        }

        public void OnNext(ActorEvent value)
        {
            _func(value);
        }
    }

}
